diff --git a/docs/architecture.mdx b/docs/architecture.mdx index dd24eb23d..3cd92d2b8 100644 --- a/docs/architecture.mdx +++ b/docs/architecture.mdx @@ -254,6 +254,25 @@ and its broker-owned service metadata over pinned cluster mTLS on the same port. The public preview still exposes hostname, hardware inventory, and utilization to devices on the subnet; do not run PAIR where that disclosure is unacceptable. +### Authenticated Peer Address Roaming + +Cluster-manager keeps at most eight local address hints per member in +`cluster/members.json` (0600), ordered by authenticated success, retaining the +pairing-time endpoint as a fallback. Only a pinned roster connection's source +address, that peer's own `selfAddresses` report, or its successful pinned +response can teach that member's record. Discovery, previews, names and relayed +addresses cannot teach it. Learned hints expire after 30 days without success; +new/unproven endpoints are dial-budgeted to one per member per 30 seconds, +including across restarts. IPv6 scopes are local to the observed connection. + +An address is never an identity: every connection checks the exact current +certificate pin and redirects are refused. Removal clears the member's hints, +including against an in-flight read. Saved manual addresses retain user +precedence; cluster-derived entries follow the authenticated reached address in +the existing node detail. If both peers lose every known path, an explicit +reachable rendezvous is still required. See +[cluster-manager](../services/nvpair-cluster-manager/README.md#peer-address-roaming). + ### Two Personalities on One Port The proxies and `nvpair-cluster-manager` serve two different things on a single diff --git a/services/nvpair-cluster-manager/README.md b/services/nvpair-cluster-manager/README.md index f360a5f4c..eb5315337 100644 --- a/services/nvpair-cluster-manager/README.md +++ b/services/nvpair-cluster-manager/README.md @@ -363,6 +363,48 @@ cleaning or renaming it, and without touching disk or announcing a change. Broker-supplied address always wins, so manual-only nodes are still invitable and discovery-less networks still work. +## Peer address roaming + +`POST /v1/cluster/roster` adds optional `selfAddresses: string[]`: numeric +`host:port` endpoints for the **authenticated sender itself**, including routed +overlay addresses from the local interface publisher. Older peers may omit it. +The receiver learns only from that pinned connection's source host (keeping the +known listening port), the sender's self-report, or a successful pinned response. +Relayed roster addresses, names, mDNS and public previews never update this set. +Identity always comes from the exact certificate pin, not an address. + +Local-only `pairingAddress`, `addressCandidates` (endpoint, first learned time, +last authenticated success, last attempt), and `nextAddressProbe` are persisted +atomically inside `cluster/members.json` with mode **0600**. These fields are not +added to the JSON-RPC member projection. The existing `ipAddress` is the most +recent authenticated reached host; the manual-node worker follows it for +cluster-derived entries, so existing node detail shows the reached address. +User-saved entries retain their address, name, kind and precedence. + +At most **8** distinct endpoints are retained per member, most recent successful +first, with the pairing-time endpoint retained as a fallback. Learned endpoints +expire after **30 days** without authenticated success (untried claims age from +first receipt; repeating a claim does not renew it). Only numeric unicast IPs +and ports 1–65535 are accepted; unspecified, multicast, broadcast, IPv4 link-local +and self-reported loopback addresses are rejected. IPv6 link-local hints use the +receiving/dialing connection's **local** zone, never the sender's interface name. + +The existing reachability chooser orders attempts within the whole-request +budget; TCP acceptance alone never proves identity. At most **one unproven +candidate per member per 30 seconds** is admitted to a dial walk, across +concurrent reconciles and restarts, rotating fairly across untried hints. +Redirects are not followed. Every application connection still verifies the +member's exact current pin. A hostile address can fail, not impersonate a peer. +Removal/rejoin is serialized with learning and clears all hints; a request or +response already in flight cannot restore removed state. + +At least one authenticated rendezvous path must remain: for example a peer +reaches the other's saved overlay address and teaches its own current addresses. +If both machines lose every known path, re-establish reachability explicitly; +this is not an unauthenticated discovery or naming service. Endorsements still +fan out certificate trust, but a third party's relayed address grants no +address-learning authority. + ## Cluster trust fan-out Pairing is point-to-point, but trust is **not** left point-to-point: every diff --git a/services/nvpair-cluster-manager/addresses.go b/services/nvpair-cluster-manager/addresses.go new file mode 100644 index 000000000..56a6f00e6 --- /dev/null +++ b/services/nvpair-cluster-manager/addresses.go @@ -0,0 +1,283 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "log" + "net" + "net/netip" + "slices" + "sort" + "strconv" + "strings" + "time" + + "nvpair-shared/netpick" +) + +const ( + maxPeerAddresses = 8 // includes the permanent pairing-time fallback + peerAddressTTL = 30 * 24 * time.Hour + newAddressInterval = 30 * time.Second +) + +// Address hints are local durable state, not membership authority and not part +// of ClusterNode's JSON-RPC projection. Only cluster-manager writes them. +type addressCandidate struct { + Address string `json:"address"` + LearnedAt int64 `json:"learnedAt"` + LastSuccess int64 `json:"lastSuccess,omitempty"` + LastAttempt int64 `json:"lastAttempt,omitempty"` +} + +type memberAddresses struct { + PairingAddress string `json:"pairingAddress,omitempty"` + AddressCandidates []addressCandidate `json:"addressCandidates,omitempty"` + NextAddressProbe int64 `json:"nextAddressProbe,omitempty"` +} + +type persistedMember struct { + ClusterNode + memberAddresses +} + +// candidateEndpoint accepts only numeric unicast addresses, never a hostname +// whose subsequent resolution could redirect a learned hint. A zone is local +// routing state: accept it only on IPv6 link-local addresses. Self-reported +// zones are replaced with the receiving connection's zone before validation. +func candidateEndpoint(endpoint string, allowLoopback bool) (string, bool) { + host, p, err := net.SplitHostPort(endpoint) + if err != nil { + return "", false + } + port, err := strconv.Atoi(p) + if err != nil || port < 1 || port > 65535 { + return "", false + } + ip, err := netip.ParseAddr(host) + if err != nil { + return "", false + } + ip = ip.Unmap() + if ip.IsUnspecified() || ip.IsMulticast() || (!ip.IsGlobalUnicast() && !ip.IsLinkLocalUnicast() && !ip.IsLoopback()) || (!allowLoopback && ip.IsLoopback()) { + return "", false + } + if ip.Is4() { + b := ip.As4() + if b[0] == 0 || b[0] >= 224 || ip.IsLinkLocalUnicast() { + return "", false + } + } + if ip.Is6() && ip.IsLinkLocalUnicast() && ip.Zone() == "" { + return "", false + } + if zone := ip.Zone(); zone != "" { + if !ip.Is6() || !ip.IsLinkLocalUnicast() || len(zone) > 64 || strings.ContainsAny(zone, "%/\\?#[]: \t\r\n") { + return "", false + } + } + return net.JoinHostPort(ip.String(), strconv.Itoa(port)), true +} + +func normalizeMemberAddresses(n *ClusterNode, now time.Time) { + a := &n.addresses + if a.PairingAddress == "" && len(a.AddressCandidates) == 0 { + a.PairingAddress = joinHostPort(n.IPAddress, n.Port, n.Port) + } + if addr, ok := candidateEndpoint(a.PairingAddress, true); ok { + a.PairingAddress = addr + } else { + a.PairingAddress = "" + } + seen := make(map[string]bool) + kept := make([]addressCandidate, 0, maxPeerAddresses-1) + for _, c := range a.AddressCandidates { + addr, ok := candidateEndpoint(c.Address, true) + if !ok || seen[addr] || max(c.LearnedAt, c.LastSuccess) < now.Add(-peerAddressTTL).UnixMilli() { + continue + } + seen[addr] = true + c.Address = addr + // Disk state is also bounded and sanitized, including unreasonable clocks. + c.LearnedAt = min(c.LearnedAt, now.UnixMilli()) + c.LastSuccess = min(c.LastSuccess, now.UnixMilli()) + kept = append(kept, c) + } + sort.SliceStable(kept, func(i, j int) bool { return kept[i].LastSuccess > kept[j].LastSuccess }) + a.AddressCandidates = kept[:min(len(kept), maxPeerAddresses-1)] + // A stale display/dial address must not bypass candidate expiry or the cap. + current := joinHostPort(n.IPAddress, n.Port, n.Port) + if current != a.PairingAddress && !slices.ContainsFunc(a.AddressCandidates, func(c addressCandidate) bool { return c.Address == current && c.LastSuccess > 0 }) { + addr := a.PairingAddress + if len(a.AddressCandidates) > 0 && a.AddressCandidates[0].LastSuccess > 0 { + addr = a.AddressCandidates[0].Address + } + if addr != "" { + n.IPAddress, n.Port = splitAddr(addr, n.Port) + } else { + n.IPAddress = "" + } + } + a.NextAddressProbe = min(a.NextAddressProbe, now.Add(newAddressInterval).UnixMilli()) +} + +func memberCandidateAddrs(n ClusterNode, now time.Time) []string { + normalizeMemberAddresses(&n, now) + out := make([]string, 0, maxPeerAddresses) + for _, c := range n.addresses.AddressCandidates { + out = append(out, c.Address) + } + if n.addresses.PairingAddress != "" && !slices.Contains(out, n.addresses.PairingAddress) { + out = append(out, n.addresses.PairingAddress) + } + return out +} + +// learnPeerAddressesLocked is called only after rechecking the exact connection +// pin and admission under rosterMu. An authenticated sender can teach only its +// own member record. No removed member is created, and no mDNS/preview/gossip +// address reaches this function. observed carries the receiver's IPv6 zone. +func (m *Manager) learnPeerAddressesLocked(uuid, observed string, reports []string, successAddr string, now time.Time) bool { + pin, pinned := m.trust.Get(uuid) + if !pinned || m.teardownPending.Load() { + return false + } + m.memMu.Lock() + defer m.memMu.Unlock() + n, ok := m.members[uuid] + if !ok || n.State != stateMember || n.AdmissionEpoch != pin.AdmissionEpoch || n.ClusterID != pin.ClusterID { + return false + } + normalizeMemberAddresses(n, now) + before := *cloneClusterNode(n) + add := func(addr string, success bool) { + if addr == n.addresses.PairingAddress && !success { + return + } + i := slices.IndexFunc(n.addresses.AddressCandidates, func(c addressCandidate) bool { return c.Address == addr }) + if i >= 0 { + if !success { + return + } // a repeated claim never renews expiry or retry budget + c := n.addresses.AddressCandidates[i] + n.addresses.AddressCandidates = append(n.addresses.AddressCandidates[:i], n.addresses.AddressCandidates[i+1:]...) + c.LastSuccess = now.UnixMilli() + n.addresses.AddressCandidates = append([]addressCandidate{c}, n.addresses.AddressCandidates...) + } else { + c := addressCandidate{Address: addr, LearnedAt: now.UnixMilli()} + if success { + c.LastSuccess = now.UnixMilli() + n.addresses.AddressCandidates = append([]addressCandidate{c}, n.addresses.AddressCandidates...) + } else if len(n.addresses.AddressCandidates) < maxPeerAddresses-1 { + n.addresses.AddressCandidates = append(n.addresses.AddressCandidates, c) + } + } + n.addresses.AddressCandidates = n.addresses.AddressCandidates[:min(len(n.addresses.AddressCandidates), maxPeerAddresses-1)] + if success { + n.IPAddress, n.Port = splitAddr(addr, n.Port) + } + } + if observed != "" { + if addr, ok := candidateEndpoint(joinHostPort(observed, n.Port, m.port), true); ok { + add(addr, true) + } + } + if addr, ok := candidateEndpoint(successAddr, true); ok { + add(addr, true) + } + // Bound processing as well as storage; a large body cannot churn the set. + via := observed + if via == "" { + via = hostOnly(successAddr) + } + for _, addr := range reports[:min(len(reports), maxPeerAddresses)] { + addr = localScope(addr, via) + if addr, ok := candidateEndpoint(addr, false); ok { + add(addr, false) + } + } + return before.IPAddress != n.IPAddress || before.Port != n.Port || !slices.Equal(before.addresses.AddressCandidates, n.addresses.AddressCandidates) +} + +// limitCandidateDials reserves at most one unproven endpoint per member per +// 30 seconds, shared by concurrent reconciles and persisted across restarts. +// A peer rotating its claims cannot reset this per-member budget. Previously +// authenticated sources/successes and the pairing fallback remain available. +func (m *Manager) limitCandidateDials(uuid string, addrs []string, now time.Time) ([]string, bool) { + m.memMu.Lock() + defer m.memMu.Unlock() + n, ok := m.members[uuid] + if !ok { + return nil, false + } + normalizeMemberAddresses(n, now) + allowed := make(map[string]bool) + for _, c := range n.addresses.AddressCandidates { + if c.LastSuccess > 0 { + allowed[c.Address] = true + } + } + allowed[n.addresses.PairingAddress] = true + var probe string + if now.UnixMilli() >= n.addresses.NextAddressProbe { + // Oldest attempted hint first prevents a dead first claim starving the rest. + oldest := int64(1<<63 - 1) + for _, c := range n.addresses.AddressCandidates { + if c.LastSuccess == 0 && slices.Contains(addrs, c.Address) && c.LastAttempt < oldest { + probe = c.Address + oldest = c.LastAttempt + } + } + if probe != "" { + n.addresses.NextAddressProbe = now.Add(newAddressInterval).UnixMilli() + for i := range n.addresses.AddressCandidates { + if n.addresses.AddressCandidates[i].Address == probe { + n.addresses.AddressCandidates[i].LastAttempt = now.UnixMilli() + } + } + allowed[probe] = true + } + } + out := make([]string, 0, maxPeerAddresses) + for _, addr := range addrs { + if allowed[addr] && !slices.Contains(out, addr) { + out = append(out, addr) + } + if len(out) == maxPeerAddresses { + break + } + } + return out, probe != "" +} + +// A reservation must reach disk before a new endpoint is dialed. Otherwise +// repeated restarts during a write failure could bypass the per-member budget. +func (m *Manager) reservePeerDialCandidates(uuid string, addrs []string) []string { + m.rosterMu.Lock() + defer m.rosterMu.Unlock() + allowed, reserved := m.limitCandidateDials(uuid, addrs, time.Now()) + if reserved { + if err := m.persistMembersErr(); err != nil { + log.Printf("persist address dial budget: %v", err) + // The in-memory reservation is retained. A second filter excludes + // unproven hints while leaving proven endpoints/fallback available. + allowed, _ = m.limitCandidateDials(uuid, allowed, time.Now()) + } + } + return allowed +} + +func localRosterAddresses(port int) []string { + var out []string + // The shared publisher includes routed overlay addresses behind the LAN. + for _, host := range netpick.LocalCandidates(netpick.Evidence{}) { + if addr, ok := candidateEndpoint(net.JoinHostPort(host, strconv.Itoa(port)), false); ok { + out = append(out, addr) + } + if len(out) == maxPeerAddresses { + break + } + } + return out +} diff --git a/services/nvpair-cluster-manager/addresses_test.go b/services/nvpair-cluster-manager/addresses_test.go new file mode 100644 index 000000000..4a9aad179 --- /dev/null +++ b/services/nvpair-cluster-manager/addresses_test.go @@ -0,0 +1,151 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "crypto/tls" + "crypto/x509" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "slices" + "strings" + "testing" + "time" +) + +func TestRoamingCandidateValidationAndIPv6Zones(t *testing.T) { + for _, addr := range []string{"0.0.0.0:1", "[::]:1", "255.255.255.255:1", "224.0.0.1:1", "[ff02::1%en0]:1", "[fe80::1]:1", "[2001:db8::1%en0]:1", "[fe80::1%bad/name]:1", "127.0.0.1:1", "169.254.1.2:1", "example.test:1", "10.0.0.1:65536", "10.0.0.1:0"} { + if got, ok := candidateEndpoint(addr, false); ok { + t.Errorf("accepted %q as %q", addr, got) + } + } + for _, addr := range []string{"100.64.0.1:14321", "192.168.1.1:14321", "[2001:db8::1]:14321", "[fe80::1%en0]:14321"} { + if got, ok := candidateEndpoint(addr, false); !ok || got != addr { + t.Errorf("rejected/changed %q: %q %v", addr, got, ok) + } + } + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingRequest(t, m, b, "[fe80::2%local0]:50000", []string{"[fe80::3%remote0]:14999"}) + n, _ := m.memberByNodeID(b.identity.NodeUUID) + got := m.resolvePeerAddrs(*n) + if !slices.Contains(got, "[fe80::3%local0]:14999") || slices.Contains(got, "[fe80::3%remote0]:14999") { + t.Fatalf("remote scope leaked into dial: %v", got) + } + if !strings.Contains(peerURL("https", "[fe80::3%local0]:14999", rosterPath), "%25local0") { + t.Fatal("IPv6 zone not URL escaped") + } +} + +func TestRoamingNewCandidateDialBudgetSurvivesClaimChurnAndReload(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + hints := []string{"100.64.0.1:14999", "100.64.0.2:14999", "100.64.0.3:14999"} + roamingRequest(t, m, b, "10.0.0.1:50000", hints) + now := time.Now() + n, _ := m.memberByNodeID(b.identity.NodeUUID) + first, reserved := m.limitCandidateDials(n.NodeUUID, m.resolvePeerAddrs(*n), now) + if !reserved || len(first) != 3 || !slices.Contains(first, hints[0]) { + t.Fatalf("first probe: %v %v", first, reserved) + } + m.persistMembers() + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + roamingRequest(t, m, b, "10.0.0.1:50000", []string{"100.64.0.4:14999"}) + n, _ = m.memberByNodeID(b.identity.NodeUUID) + second, reserved := m.limitCandidateDials(n.NodeUUID, m.resolvePeerAddrs(*n), now.Add(time.Second)) + if reserved || len(second) != 2 { + t.Fatalf("new claims/restart reset rate limit: %v %v", second, reserved) + } + third, reserved := m.limitCandidateDials(n.NodeUUID, m.resolvePeerAddrs(*n), now.Add(31*time.Second)) + if !reserved || !slices.Contains(third, hints[1]) { + t.Fatalf("untried hint starved: %v %v", third, reserved) + } +} + +type roamingBlockedBody struct { + entered, release chan struct{} + reader io.Reader +} + +func (b *roamingBlockedBody) Read(p []byte) (int, error) { + select { + case <-b.entered: + default: + close(b.entered) + } + <-b.release + return b.reader.Read(p) +} +func (b *roamingBlockedBody) Close() error { return nil } + +func TestRoamingInFlightRequestCannotRestoreRemovedMember(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + blocked := &roamingBlockedBody{entered: make(chan struct{}), release: make(chan struct{}), reader: strings.NewReader(`{"clusterId":"cluster-1","selfAddresses":["100.64.0.8:14999"]}`)} + r := httptest.NewRequest(http.MethodPost, rosterPath, nil) + r.Body = blocked + r.RemoteAddr = "10.0.0.8:50000" + cert, err := x509.ParseCertificate(b.identity.Cert.Certificate[0]) + if err != nil { + t.Fatal(err) + } + r.TLS = &tls.ConnectionState{PeerCertificates: []*x509.Certificate{cert}} + w := httptest.NewRecorder() + done := make(chan struct{}) + go func() { defer close(done); m.handleRoster(w, r) }() + <-blocked.entered + m.rosterMu.Lock() + if err := m.trust.Remove(b.identity.NodeUUID); err != nil { + t.Fatal(err) + } + m.removeMemberByUUID(b.identity.NodeUUID) + m.persistMembers() + m.rosterMu.Unlock() + close(blocked.release) + <-done + if w.Code != http.StatusForbidden { + t.Fatalf("stale request accepted: %d", w.Code) + } + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + if _, exists := m.memberByNodeID(b.identity.NodeUUID); exists { + t.Fatal("in-flight learning resurrected removed member") + } +} + +func TestRoamingLearnsSelfReportFromPinnedResponseOnly(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + const hint = "100.64.0.8:14999" + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if _, ok := b.verifyClientPin(r); !ok { + http.Error(w, "forbidden", 403) + return + } + _ = json.NewEncoder(w).Encode(map[string]interface{}{"clusterId": "cluster-1", "selfAddresses": []string{hint}}) + })) + pinTrusted(t, b, m.identity.NodeUUID, string(m.identity.CertPEM), m.identity.CertFingerprint) + srv.TLS = b.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "https://") + host, port := splitAddr(addr, m.port) + m.upsertMember(&ClusterNode{NodeUUID: b.identity.NodeUUID, ID: "peer", IPAddress: host, Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + outcome, _ := m.reconcileWithin(context.Background(), []string{addr}, b.identity.NodeUUID) + if outcome != reconcileAccepted { + t.Fatalf("outcome=%v", outcome) + } + n, _ := m.memberByNodeID(b.identity.NodeUUID) + if !slices.Contains(m.resolvePeerAddrs(*n), hint) { + t.Fatal("pinned response self-report not learned") + } +} diff --git a/services/nvpair-cluster-manager/connreuse_test.go b/services/nvpair-cluster-manager/connreuse_test.go index 83254c579..354b7b9f7 100644 --- a/services/nvpair-cluster-manager/connreuse_test.go +++ b/services/nvpair-cluster-manager/connreuse_test.go @@ -64,6 +64,8 @@ func TestReconcile_ReusesPeerConnections(t *testing.T) { defer ts.Close() addr := ts.Listener.Addr().String() + host, port := splitAddr(addr, m.port) + m.upsertMember(&ClusterNode{NodeUUID: peerUUID, ID: "peer", IPAddress: host, Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) const rounds = 5 for i := 0; i < rounds; i++ { if outcome, _ := m.reconcileWith([]string{addr}, peerUUID); outcome != reconcileAccepted { diff --git a/services/nvpair-cluster-manager/httpserver.go b/services/nvpair-cluster-manager/httpserver.go index 7eb71ec5d..f59eb1e0c 100644 --- a/services/nvpair-cluster-manager/httpserver.go +++ b/services/nvpair-cluster-manager/httpserver.go @@ -507,16 +507,8 @@ func (m *Manager) prepareJoinerInitialComplete(inviteID string, sess *pairingSes log.Printf("pairing %s: cannot read inviter PairingInfo: %v", inviteID, err) return nil } - // The inviter's advertised link-local address carries the zone of its own - // interface; give it ours before it is dialled or recorded as the member's - // address, and keep it as an observation so a roster claim of the inviter's - // spelling cannot replace it. - if scoped := localScope(pi.Addr, inviterIP); scoped != pi.Addr { - pi.Addr = scoped - if host, _, err := net.SplitHostPort(scoped); err == nil { - m.noteObservedPeerHost(pi.NodeUUID, host) - } - } + // Remote zones name the peer's interfaces; scope to the incoming link. + pi.Addr = localScope(pi.Addr, inviterIP) sess.peerPairing = pi sess.peerCert = cert sess.addr = pi.Addr @@ -656,7 +648,6 @@ func (m *Manager) onInviterPaired(inviteID string, sess *pairingSession) error { host, port := splitAddr(pi.Addr, m.port) if sess.peerSrcIP != "" { host = sess.peerSrcIP - m.noteObservedPeerHost(pi.NodeUUID, host) } if host != "" { pi.Addr = net.JoinHostPort(host, strconv.Itoa(port)) diff --git a/services/nvpair-cluster-manager/manager.go b/services/nvpair-cluster-manager/manager.go index bef58bd04..9dc3819ab 100644 --- a/services/nvpair-cluster-manager/manager.go +++ b/services/nvpair-cluster-manager/manager.go @@ -82,9 +82,10 @@ type Manager struct { // wait is never erased by a now-stale unanimous-403 verdict. clusterGen atomic.Uint64 - memMu sync.Mutex - members map[string]*ClusterNode // keyed by nodeUuid - invites map[string]*Invite // keyed by inviteId + membersWriteMu sync.Mutex // serialize durable snapshots, including address learning + memMu sync.Mutex + members map[string]*ClusterNode // keyed by nodeUuid + invites map[string]*Invite // keyed by inviteId tsMu sync.Mutex tombstones map[string]Tombstone // keyed by removed nodeUuid; newest wins (fan-out removal) @@ -132,13 +133,6 @@ type Manager struct { browser *Browser - // observedPeerHosts maps a member's uuid to the source host it was last seen - // connecting from over an authenticated transport. Memory-only: it is - // re-earned by the first inbound reconcile after a restart, and a stale - // persisted copy would be a claim rather than an observation. - observedMu sync.Mutex - observedPeerHosts map[string]string - // peerAddrs remembers which of a member's advertised addresses answers, so a // multi-homed peer is reconciled with at an address that works rather than at // whichever one it happens to rank first. diff --git a/services/nvpair-cluster-manager/membership.go b/services/nvpair-cluster-manager/membership.go index d207ab792..00ac48b5a 100644 --- a/services/nvpair-cluster-manager/membership.go +++ b/services/nvpair-cluster-manager/membership.go @@ -61,6 +61,8 @@ type ClusterNode struct { State MembershipState `json:"state"` JoinedAt *int64 `json:"joinedAt"` LastSeen *int64 `json:"lastSeen"` + + addresses memberAddresses // local-only durable hints; never serialized to IPC } // Invite is the Broker-facing view of a pairing session (§6). Certificates are @@ -88,6 +90,7 @@ func cloneClusterNode(node *ClusterNode) *ClusterNode { return nil } cp := *node + cp.addresses.AddressCandidates = append([]addressCandidate(nil), node.addresses.AddressCandidates...) if node.JoinedAt != nil { v := *node.JoinedAt cp.JoinedAt = &v @@ -122,7 +125,11 @@ func cloneInvite(inv *Invite) *Invite { // upsertMember inserts or replaces a member, keyed by nodeUuid. func (m *Manager) upsertMember(node *ClusterNode) { m.memMu.Lock() - m.members[node.NodeUUID] = cloneClusterNode(node) + copy := cloneClusterNode(node) + if copy.State == stateMember { + normalizeMemberAddresses(copy, time.Now()) + } + m.members[node.NodeUUID] = copy m.memMu.Unlock() m.clusterGen.Add(1) } @@ -133,6 +140,7 @@ func (m *Manager) removeMemberByUUID(uuid string) bool { _, ok := m.members[uuid] delete(m.members, uuid) m.memMu.Unlock() + m.peerAddrs.Forget(uuid) if ok { m.clusterGen.Add(1) } @@ -142,8 +150,12 @@ func (m *Manager) removeMemberByUUID(uuid string) bool { // resetMembership clears the entire member set (used when leaving a cluster). func (m *Manager) resetMembership() { m.memMu.Lock() + old := m.members m.members = make(map[string]*ClusterNode) m.memMu.Unlock() + for uuid := range old { + m.peerAddrs.Forget(uuid) + } m.clusterGen.Add(1) } @@ -179,68 +191,6 @@ func (m *Manager) memberByNodeID(idOrUUID string) (*ClusterNode, bool) { return nil, false } -// setMemberAddr updates a member's reachable host:port under the membership -// lock, returning whether anything actually changed. A zero/invalid port keeps -// the member's existing listening port (falling back to the local default), so -// callers learning only a host (e.g. a TCP source IP, whose port is ephemeral) -// don't clobber the real listening port. Callers persist and emit -// nodes:changed when this returns true. -func (m *Manager) setMemberAddr(uuid, host string, port int) bool { - if host == "" { - return false - } - m.memMu.Lock() - defer m.memMu.Unlock() - n, ok := m.members[uuid] - if !ok { - return false - } - if port < 1 || port > 65535 { - port = n.Port - } - if port < 1 || port > 65535 { - port = m.port - } - if n.IPAddress == host && n.Port == port { - return false - } - n.IPAddress = host - n.Port = port - return true -} - -// noteObservedPeerHost records the source host an authenticated peer connected -// from and applies it as that member's address, returning whether anything -// changed. -// -// An address observed on a connection is the only claim about a peer's -// reachability this node has verified: the peer reached us from it, over a -// transport pinned to its certificate. A peer's own advertised address — or worse, -// a third peer's relayed copy of it — is that peer guessing which of its addresses -// works from here, which is exactly the guess that goes wrong on a multi-homed -// host. So the observation is remembered, and refreshMemberAddr will not overwrite -// it with a claim. -func (m *Manager) noteObservedPeerHost(uuid, host string) bool { - if uuid == "" || host == "" { - return false - } - m.observedMu.Lock() - if m.observedPeerHosts == nil { - m.observedPeerHosts = make(map[string]string) - } - m.observedPeerHosts[uuid] = host - m.observedMu.Unlock() - return m.setMemberAddr(uuid, host, 0) -} - -// observedPeerHost returns the source host this peer was last seen connecting -// from, or "" if it never has been. -func (m *Manager) observedPeerHost(uuid string) string { - m.observedMu.Lock() - defer m.observedMu.Unlock() - return m.observedPeerHosts[uuid] -} - // updateMemberIdentity refreshes a member's display fields (id/name) in place, // keyed by its stable nodeUuid. Empty inputs are ignored so a partial update // can't blank a known value. Returns whether anything changed. Display-only: the diff --git a/services/nvpair-cluster-manager/membership_store.go b/services/nvpair-cluster-manager/membership_store.go index f5ac30ab3..6c44f565e 100644 --- a/services/nvpair-cluster-manager/membership_store.go +++ b/services/nvpair-cluster-manager/membership_store.go @@ -11,6 +11,7 @@ import ( "os" "path/filepath" "sort" + "time" ) // membersFile is the durable, lean roster persisted under the cluster dir. Only @@ -30,11 +31,15 @@ func (m *Manager) persistMembers() { } func (m *Manager) persistMembersErr() error { + m.membersWriteMu.Lock() + defer m.membersWriteMu.Unlock() m.memMu.Lock() - durable := make([]ClusterNode, 0, len(m.members)) + durable := make([]persistedMember, 0, len(m.members)) for _, n := range m.members { if n.State == stateMember { - durable = append(durable, *cloneClusterNode(n)) + normalizeMemberAddresses(n, time.Now()) + copy := cloneClusterNode(n) + durable = append(durable, persistedMember{ClusterNode: *copy, memberAddresses: copy.addresses}) } } m.memMu.Unlock() @@ -57,14 +62,16 @@ func (m *Manager) loadMembers() error { if err != nil { return err } - var nodes []ClusterNode + var nodes []persistedMember if err := json.Unmarshal(data, &nodes); err != nil { return err } m.memMu.Lock() defer m.memMu.Unlock() for i := range nodes { - n := nodes[i] + n := nodes[i].ClusterNode + n.addresses = nodes[i].memberAddresses + normalizeMemberAddresses(&n, time.Now()) m.members[n.NodeUUID] = &n } return nil diff --git a/services/nvpair-cluster-manager/mtls.go b/services/nvpair-cluster-manager/mtls.go index 43d6cca3c..e834bd864 100644 --- a/services/nvpair-cluster-manager/mtls.go +++ b/services/nvpair-cluster-manager/mtls.go @@ -50,7 +50,11 @@ func (m *Manager) peerClient(peerUUID string) (*http.Client, error) { if !ok { return nil, fmt.Errorf("no pin for peer %s", peerUUID) } - return client, nil + // Peer responses cannot send candidate dials to a different endpoint or + // downgrade transport. Keep the pooled transport, but prohibit redirects. + bounded := *client + bounded.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } + return &bounded, nil } func (m *Manager) buildPairingClientTLSConfig(peerDER []byte) (*tls.Config, error) { @@ -152,20 +156,35 @@ func (m *Manager) handleMembersRemove(w http.ResponseWriter, r *http.Request) { // notifyPeerRemoval best-effort tells a peer it has been removed, over mTLS, // while the peer's pin still exists (call before deleting it locally). -func (m *Manager) notifyPeerRemoval(addr, peerUUID string, proof RemovalProof) { +func (m *Manager) notifyPeerRemoval(node ClusterNode, proof RemovalProof) { + peerUUID := node.NodeUUID client, err := m.peerClient(peerUUID) if err != nil { log.Printf("notify removal: %v", err) return } + ctx, cancel := context.WithTimeout(context.Background(), pairingHTTPTimeout) + defer cancel() + addrs := m.reservePeerDialCandidates(peerUUID, m.resolvePeerAddrs(node)) body, _ := json.Marshal(membersRemoveRequest{NodeUUID: m.identity.NodeUUID, Proof: proof}) - resp, err := client.Post(peerURL("https", addr, membersRemovePath), "application/json", bytes.NewReader(body)) - if err != nil { - log.Printf("notify removal to %s (%s): %v", addr, peerUUID, err) + for _, addr := range confirmedFirst(addrs, m.peerAddrs.ChooseWithin(ctx, peerUUID, addrs)) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, peerURL("https", addr, membersRemovePath), bytes.NewReader(body)) + if err != nil { + continue + } + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + m.peerAddrs.Forget(peerUUID) + if ctx.Err() != nil { + break + } + continue + } + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBodyBytes)) + _ = resp.Body.Close() return } - _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBodyBytes)) - _ = resp.Body.Close() } // serveSplit serves both transport modes on one listener via the shared diff --git a/services/nvpair-cluster-manager/peer_addr_test.go b/services/nvpair-cluster-manager/peer_addr_test.go index 024aff153..12166ba40 100644 --- a/services/nvpair-cluster-manager/peer_addr_test.go +++ b/services/nvpair-cluster-manager/peer_addr_test.go @@ -14,8 +14,6 @@ import ( "sync/atomic" "testing" "time" - - "nvpair-shared/noderec" ) // unreachableAddr is in TEST-NET-1 (RFC 5737), which is reserved for @@ -27,22 +25,22 @@ const unreachableAddr = "192.0.2.1:14321" // could not reach, so every reconcile burned the full request timeout against it // while the address that worked sat unused in the same candidate list. func TestReconcileWithFailsOverToAnAnsweringAddress(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - mA := newTestManagerPort(t, 15031) - mB := newTestManagerPort(t, 15032) - go func() { _ = mA.runHTTP(ctx) }() - go func() { _ = mB.runHTTP(ctx) }() - time.Sleep(400 * time.Millisecond) - + mA, mB := newTestManager(t), newTestManager(t) + srv := httptest.NewUnstartedServer(http.HandlerFunc(mA.handleRoster)) + _, port := splitAddr(srv.Listener.Addr().String(), mA.port) + mA.port = port + srv.TLS = mA.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() pinTrusted(t, mA, mB.identity.NodeUUID, string(mB.identity.CertPEM), mB.identity.CertFingerprint) pinTrusted(t, mB, mA.identity.NodeUUID, string(mA.identity.CertPEM), mA.identity.CertFingerprint) - mA.upsertMember(&ClusterNode{NodeUUID: mB.identity.NodeUUID, ID: "node-b", IPAddress: "127.0.0.1", Port: 15032, AdmissionEpoch: 1, State: stateMember}) - mB.upsertMember(&ClusterNode{NodeUUID: mA.identity.NodeUUID, ID: "node-a", IPAddress: "192.0.2.1", Port: 15031, AdmissionEpoch: 1, State: stateMember}) - - working := net.JoinHostPort("127.0.0.1", strconv.Itoa(15031)) - outcome, _ := mB.reconcileWith([]string{unreachableAddr, working}, mA.identity.NodeUUID) + mA.upsertMember(&ClusterNode{NodeUUID: mB.identity.NodeUUID, ID: "node-b", IPAddress: "127.0.0.1", Port: mB.port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + mB.upsertMember(&ClusterNode{NodeUUID: mA.identity.NodeUUID, ID: "node-a", IPAddress: "192.0.2.1", Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + working := strings.TrimPrefix(srv.URL, "https://") + mB.memMu.Lock() + mB.members[mA.identity.NodeUUID].addresses.AddressCandidates = []addressCandidate{{Address: working, LearnedAt: time.Now().UnixMilli()}} + mB.memMu.Unlock() + outcome, _ := mB.reconcileWith([]string{net.JoinHostPort("192.0.2.1", strconv.Itoa(port)), working}, mA.identity.NodeUUID) if outcome != reconcileAccepted { t.Fatalf("reconcile outcome = %v, want accepted via the second candidate", outcome) } @@ -58,6 +56,12 @@ func TestReconcileBudgetCoversTheWholeCandidateWalk(t *testing.T) { pinTrusted(t, m, peerUUID, certPEM, fingerprint) blackholed := []string{"192.0.2.1:14321", "192.0.2.2:14321", "192.0.2.3:14321", "192.0.2.4:14321"} + m.upsertMember(&ClusterNode{NodeUUID: peerUUID, ID: "peer", IPAddress: "192.0.2.1", Port: 14321, State: stateMember}) + m.memMu.Lock() + for _, addr := range blackholed[1:] { + m.members[peerUUID].addresses.AddressCandidates = append(m.members[peerUUID].addresses.AddressCandidates, addressCandidate{Address: addr, LearnedAt: time.Now().UnixMilli()}) + } + m.memMu.Unlock() budget := 600 * time.Millisecond ctx, cancel := context.WithTimeout(context.Background(), budget) defer cancel() @@ -76,102 +80,7 @@ func TestReconcileBudgetCoversTheWholeCandidateWalk(t *testing.T) { } } -// TestResolvePeerAddrsPutsTheRecordedAddressFirst: the pairing-time address was -// observed working from here, so it leads — but the peer's other advertised -// addresses follow it rather than being discarded. -func TestResolvePeerAddrsPutsTheRecordedAddressFirst(t *testing.T) { - m := newTestManagerPort(t, 14321) - peer := "33333333-3333-3333-3333-333333333333" - m.browser = newBrowser() - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "192.168.240.1", - IPs: []string{"192.168.240.1", "10.172.55.129"}, - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - - got := m.resolvePeerAddrs(ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "10.172.55.129", Port: 14321}) - want := []string{"10.172.55.129:14321", "192.168.240.1:14321"} - if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { - t.Fatalf("resolvePeerAddrs = %v, want %v", got, want) - } -} - -// TestRefreshMemberAddrsFromMDNSKeepsAStillAdvertisedAddress: the stored address -// was proven from this host, while the peer's own first choice is ranked from its -// vantage point, so a peer that still advertises what we hold must not be -// redirected onto a link only its neighbour can use. -func TestRefreshMemberAddrsFromMDNSKeepsAStillAdvertisedAddress(t *testing.T) { - m := newTestManagerPort(t, 14321) - peer := "44444444-4444-4444-4444-444444444444" - m.upsertMember(&ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "10.172.55.129", Port: 14321, State: stateMember}) - m.browser = newBrowser() - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "192.168.240.1", - IPs: []string{"192.168.240.1", "10.172.55.129"}, - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "10.172.55.129" { - t.Fatalf("stored address = %q, want it kept at 10.172.55.129", n.IPAddress) - } - - // A peer that has stopped advertising the stored address really moved. - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "10.172.55.200", - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "10.172.55.200" { - t.Fatalf("stored address = %q, want the moved peer's 10.172.55.200", n.IPAddress) - } -} - -func TestResolvePeerAddrs_PrefersLocalLANOverRecordedCGNAT(t *testing.T) { - m := newTestManagerPort(t, 14321) - peer := "55555555-5555-5555-5555-555555555555" - m.browser = newBrowser() - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "192.168.1.50", - IPs: []string{"192.168.1.50", "100.64.1.2"}, - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - - // Node was recorded with Tailscale CGNAT address. - got := m.resolvePeerAddrs(ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "100.64.1.2", Port: 14321}) - want := []string{"192.168.1.50:14321", "100.64.1.2:14321"} - if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { - t.Fatalf("resolvePeerAddrs = %v, want LAN first %v", got, want) - } -} - -func TestRefreshMemberAddrsFromMDNS_UpgradesCGNATToLAN(t *testing.T) { - m := newTestManagerPort(t, 14321) - peer := "66666666-6666-6666-6666-666666666666" - // Node currently holds Tailscale CGNAT address. - m.upsertMember(&ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "100.64.1.2", Port: 14321, State: stateMember}) - m.browser = newBrowser() - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "192.168.1.50", - IPs: []string{"192.168.1.50", "100.64.1.2"}, - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "192.168.1.50" { - t.Fatalf("stored address = %q, want upgraded to LAN 192.168.1.50", n.IPAddress) - } -} +// Reachability confirmation only reorders hints; authentication remains mTLS. func TestConfirmedFirst(t *testing.T) { addrs := []string{"a:1", "b:1", "c:1"} diff --git a/services/nvpair-cluster-manager/reconnect_test.go b/services/nvpair-cluster-manager/reconnect_test.go index c60760c78..3e649ab11 100644 --- a/services/nvpair-cluster-manager/reconnect_test.go +++ b/services/nvpair-cluster-manager/reconnect_test.go @@ -11,42 +11,9 @@ import ( "time" ) -// TestSetMemberAddr covers the address-update helper: a host-only update keeps -// the stored listening port, identical updates are no-ops, an explicit port is -// honored, and unknown/empty inputs change nothing. -func TestSetMemberAddr(t *testing.T) { - m := newTestManager(t) - uuid := "11111111-1111-1111-1111-111111111111" - m.upsertMember(&ClusterNode{NodeUUID: uuid, ID: "n", IPAddress: "10.0.0.1", Port: 14321, State: stateMember}) +// The real mTLS endpoint learns only the authenticated source host and keeps +// the paired listening port, never the ephemeral client port. - if !m.setMemberAddr(uuid, "192.168.1.50", 0) { - t.Fatal("expected a change for a new host") - } - if n, _ := m.memberByNodeID(uuid); n.IPAddress != "192.168.1.50" || n.Port != 14321 { - t.Fatalf("host-only update wrong: got %s:%d want 192.168.1.50:14321", n.IPAddress, n.Port) - } - if m.setMemberAddr(uuid, "192.168.1.50", 0) { - t.Fatal("identical addr should not report a change") - } - if !m.setMemberAddr(uuid, "192.168.1.50", 14999) { - t.Fatal("expected a change for a new port") - } - if n, _ := m.memberByNodeID(uuid); n.Port != 14999 { - t.Fatalf("explicit port not applied: got %d want 14999", n.Port) - } - if m.setMemberAddr("does-not-exist", "1.2.3.4", 0) { - t.Fatal("unknown uuid should not report a change") - } - if m.setMemberAddr(uuid, "", 0) { - t.Fatal("empty host should not report a change") - } -} - -// TestHandleRosterLearnsSourceAddr drives the real mTLS roster endpoint over -// loopback: the server holds a STALE address for the client (as if the client -// moved since pairing), and after one reconcile it must have learned the -// client's real source IP from the authenticated connection while preserving -// the client's known listening port. func TestHandleRosterLearnsSourceAddr(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -61,8 +28,8 @@ func TestHandleRosterLearnsSourceAddr(t *testing.T) { pinTrusted(t, mB, mA.identity.NodeUUID, string(mA.identity.CertPEM), mA.identity.CertFingerprint) // mA's record for mB points at a stale IP; mB still reaches mA fine. - mA.upsertMember(&ClusterNode{NodeUUID: mB.identity.NodeUUID, ID: "node-b", IPAddress: "10.9.9.9", Port: 15022, State: stateMember}) - mB.upsertMember(&ClusterNode{NodeUUID: mA.identity.NodeUUID, ID: "node-a", IPAddress: "127.0.0.1", Port: 15021, State: stateMember}) + mA.upsertMember(&ClusterNode{NodeUUID: mB.identity.NodeUUID, ID: "node-b", IPAddress: "10.9.9.9", Port: 15022, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + mB.upsertMember(&ClusterNode{NodeUUID: mA.identity.NodeUUID, ID: "node-a", IPAddress: "127.0.0.1", Port: 15021, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) mB.reconcileWith([]string{net.JoinHostPort("127.0.0.1", strconv.Itoa(15021))}, mA.identity.NodeUUID) @@ -81,29 +48,3 @@ func TestHandleRosterLearnsSourceAddr(t *testing.T) { // TestRefreshMemberAddrsFromMDNS verifies the discovery backstop: a member // whose stored address is stale is refreshed from the mDNS browse map, the // pass is a safe no-op without a browser, and the self entry is never rewritten. -func TestRefreshMemberAddrsFromMDNS(t *testing.T) { - m := newTestManager(t) - peer := "22222222-2222-2222-2222-222222222222" - m.upsertMember(&ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "10.9.9.9", Port: 14321, State: stateMember}) - - // No browser configured yet: must not panic and must change nothing. - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "10.9.9.9" { - t.Fatalf("addr changed without a browser: got %q", n.IPAddress) - } - - m.browser = newBrowser() - m.browser.seed(peer, "192.168.1.77", 14321) - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "192.168.1.77" { - t.Fatalf("mDNS refresh did not update stored addr: got %q want 192.168.1.77", n.IPAddress) - } - - // The self entry must never be redirected by (untrusted) mDNS. - m.addSelfMember() - m.browser.seed(m.identity.NodeUUID, "5.5.5.5", 14321) - m.refreshMemberAddrsFromMDNS() - if self, _ := m.memberByNodeID(m.identity.NodeUUID); self.IPAddress != "127.0.0.1" { - t.Fatalf("self address must not be rewritten from mDNS: got %q", self.IPAddress) - } -} diff --git a/services/nvpair-cluster-manager/remove.go b/services/nvpair-cluster-manager/remove.go index 80b604660..1d703dbaa 100644 --- a/services/nvpair-cluster-manager/remove.go +++ b/services/nvpair-cluster-manager/remove.go @@ -6,8 +6,6 @@ package main import ( "encoding/json" "fmt" - "net" - "strconv" ) type removeParams struct { @@ -92,9 +90,7 @@ func (m *Manager) handleNodesRemove(msg *Message) { } // Notify first, while the peer's pin is still present to authenticate mTLS. - if node.IPAddress != "" { - m.notifyPeerRemoval(net.JoinHostPort(node.IPAddress, strconv.Itoa(node.Port)), node.NodeUUID, proof) - } + m.notifyPeerRemoval(*node, proof) // De-pin and drop membership under the cluster-composition boundary so the // removal is atomic against a concurrent self-remove teardown — the two can't // interleave and leave a half-cleared roster. The network notify above stays diff --git a/services/nvpair-cluster-manager/roaming_order_test.go b/services/nvpair-cluster-manager/roaming_order_test.go new file mode 100644 index 000000000..1f9a71406 --- /dev/null +++ b/services/nvpair-cluster-manager/roaming_order_test.go @@ -0,0 +1,20 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "slices" + "testing" +) + +func TestRoamingSuccessfulPairingFallbackBecomesMostRecent(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingRequest(t, m, b, "10.0.0.2:50000", nil) + roamingRequest(t, m, b, "192.0.2.20:50000", nil) + n, _ := m.memberByNodeID(b.identity.NodeUUID) + if got := m.resolvePeerAddrs(*n); !slices.Equal(got, []string{"192.0.2.20:14999", "10.0.0.2:14999"}) { + t.Fatalf("most recent success not first: %v", got) + } +} diff --git a/services/nvpair-cluster-manager/roaming_persistence_test.go b/services/nvpair-cluster-manager/roaming_persistence_test.go new file mode 100644 index 000000000..60559b188 --- /dev/null +++ b/services/nvpair-cluster-manager/roaming_persistence_test.go @@ -0,0 +1,35 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "net/http" + "net/http/httptest" + "os" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestRoamingNewCandidateNeedsDurableDialBudget(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + var hits atomic.Int32 + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { hits.Add(1); w.WriteHeader(200) })) + srv.TLS = b.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "https://") + m.memMu.Lock() + m.members[b.identity.NodeUUID].addresses.AddressCandidates = []addressCandidate{{Address: addr, LearnedAt: time.Now().UnixMilli()}} + m.memMu.Unlock() + if err := os.Mkdir(m.membersPath(), 0700); err != nil { + t.Fatal(err) + } // atomic rename must fail + m.reconcileWith([]string{addr}, b.identity.NodeUUID) + if hits.Load() != 0 { + t.Fatal("dialed unproven hint without persisting its retry budget") + } +} diff --git a/services/nvpair-cluster-manager/roaming_redirect_test.go b/services/nvpair-cluster-manager/roaming_redirect_test.go new file mode 100644 index 000000000..dde48d374 --- /dev/null +++ b/services/nvpair-cluster-manager/roaming_redirect_test.go @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" +) + +func TestRoamingPinnedPeerCannotRedirectCandidateDials(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + var hits atomic.Int32 + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { hits.Add(1) })) + defer target.Close() + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL, http.StatusTemporaryRedirect) + })) + srv.TLS = b.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "https://") + host, port := splitAddr(addr, m.port) + m.upsertMember(&ClusterNode{NodeUUID: b.identity.NodeUUID, ID: "peer", IPAddress: host, Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + m.reconcileWith([]string{addr}, b.identity.NodeUUID) + if hits.Load() != 0 { + t.Fatal("pinned peer redirected a dial outside the candidate budget") + } +} diff --git a/services/nvpair-cluster-manager/roaming_test.go b/services/nvpair-cluster-manager/roaming_test.go new file mode 100644 index 000000000..4713aed1b --- /dev/null +++ b/services/nvpair-cluster-manager/roaming_test.go @@ -0,0 +1,188 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "crypto/tls" + "crypto/x509" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "slices" + "strconv" + "strings" + "testing" + "time" +) + +func roamingMember(t *testing.T, m, peer *Manager) { + t.Helper() + pinTrusted(t, m, peer.identity.NodeUUID, string(peer.identity.CertPEM), peer.identity.CertFingerprint) + m.upsertMember(&ClusterNode{ID: peer.identity.NodeID, NodeUUID: peer.identity.NodeUUID, Name: "peer", IPAddress: "192.0.2.20", Port: peer.port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) +} + +func roamingRequest(t *testing.T, m, peer *Manager, source string, hints []string) int { + t.Helper() + body, err := json.Marshal(map[string]interface{}{"clusterId": "cluster-1", "selfAddresses": hints}) + if err != nil { + t.Fatal(err) + } + r := httptest.NewRequest(http.MethodPost, rosterPath, bytes.NewReader(body)) + cert, err := x509.ParseCertificate(peer.identity.Cert.Certificate[0]) + if err != nil { + t.Fatal(err) + } + r.TLS = &tls.ConnectionState{PeerCertificates: []*x509.Certificate{cert}} + r.RemoteAddr = source + w := httptest.NewRecorder() + m.handleRoster(w, r) + return w.Code +} + +func TestRoamingSelfReportsBelongOnlyToPinnedSender(t *testing.T) { + m, b, c, unknown := newTestManager(t), newTestManager(t), newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingMember(t, m, c) + hints := []string{"100.64.0.8:14999", "[2001:db8::8]:14999"} + if code := roamingRequest(t, m, unknown, "10.0.0.8:50000", hints); code != http.StatusForbidden { + t.Fatalf("unknown cert: %d", code) + } + if code := roamingRequest(t, m, c, "10.0.0.9:50000", hints); code != http.StatusOK { + t.Fatalf("pinned cert: %d", code) + } + bn, _ := m.memberByNodeID(b.identity.NodeUUID) + if got := m.resolvePeerAddrs(*bn); !slices.Equal(got, []string{"192.0.2.20:14999"}) { + t.Fatalf("another cert taught B addresses: %v", got) + } + cn, _ := m.memberByNodeID(c.identity.NodeUUID) + got := m.resolvePeerAddrs(*cn) + if !slices.Contains(got, hints[0]) || !slices.Contains(got, hints[1]) { + t.Fatalf("verified self-report was not learned: %v", got) + } +} + +func TestRoamingCandidatesBoundedOrderedPersistedAndExpired(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + var hints []string + for i := 1; i <= 30; i++ { + hints = append(hints, "100.64.1."+strconv.Itoa(i)+":14999") + } + // Many valid endpoints, followed by the most recently observed connection. + hints = append(hints, "100.64.0.1:14999", "100.64.0.2:14999", "100.64.0.3:14999", "100.64.0.4:14999", "100.64.0.5:14999", "100.64.0.6:14999", "100.64.0.7:14999", "100.64.0.8:14999", "100.64.0.9:14999") + roamingRequest(t, m, b, "10.0.0.1:50000", hints) + roamingRequest(t, m, b, "10.0.0.2:50000", nil) + n, _ := m.memberByNodeID(b.identity.NodeUUID) + got := m.resolvePeerAddrs(*n) + if len(got) != 8 || got[0] != "10.0.0.2:14999" || got[1] != "10.0.0.1:14999" || got[len(got)-1] != "192.0.2.20:14999" { + t.Fatalf("candidate order/bound/fallback: %v", got) + } + if err := m.persistMembersErr(); err != nil { + t.Fatal(err) + } + st, err := os.Stat(m.membersPath()) + if err != nil { + t.Fatal(err) + } + if st.Mode().Perm() != 0600 { + t.Fatalf("permissions: %v", st.Mode()) + } + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + n, _ = m.memberByNodeID(b.identity.NodeUUID) + if after := m.resolvePeerAddrs(*n); !slices.Equal(after, got) { + t.Fatalf("reload: %v != %v", after, got) + } + // Age every learned hint through the actual persistence format, then reload. + data, _ := os.ReadFile(m.membersPath()) + var stored []struct { + ClusterNode + PairingAddress string `json:"pairingAddress"` + AddressCandidates []struct { + Address string `json:"address"` + LearnedAt int64 `json:"learnedAt"` + LastSuccess int64 `json:"lastSuccess"` + } `json:"addressCandidates"` + } + if err := json.Unmarshal(data, &stored); err != nil { + t.Fatal(err) + } + if len(stored) != 1 || len(stored[0].AddressCandidates) == 0 { + t.Fatal("learned candidates not persisted") + } + for i := range stored[0].AddressCandidates { + stored[0].AddressCandidates[i].LearnedAt = time.Now().Add(-31 * 24 * time.Hour).UnixMilli() + stored[0].AddressCandidates[i].LastSuccess = stored[0].AddressCandidates[i].LearnedAt + } + data, _ = json.Marshal(stored) + if err := os.WriteFile(m.membersPath(), data, 0600); err != nil { + t.Fatal(err) + } + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + n, _ = m.memberByNodeID(b.identity.NodeUUID) + if after := m.resolvePeerAddrs(*n); !slices.Equal(after, []string{"192.0.2.20:14999"}) { + t.Fatalf("expired hints survived: %v", after) + } + m.removeMemberByUUID(b.identity.NodeUUID) + m.persistMembers() + roamingRequest(t, m, b, "10.0.0.3:50000", hints) + if _, exists := m.memberByNodeID(b.identity.NodeUUID); exists { + t.Fatal("removed member resurrected by hint") + } +} + +func TestRoamingIgnoresDiscoveryAndRelayedAddresses(t *testing.T) { + m, b, c := newTestManager(t), newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingMember(t, m, c) + m.browser = newBrowser() + m.browser.seed(b.identity.NodeUUID, "10.0.0.99", 14321) + m.mergeRoster(&Roster{ClusterID: "cluster-1", Members: []RosterEntry{{NodeUUID: b.identity.NodeUUID, Addr: "10.0.0.98:14321"}}}, c.identity.NodeUUID) + n, _ := m.memberByNodeID(b.identity.NodeUUID) + if got := m.resolvePeerAddrs(*n); !slices.Equal(got, []string{"192.0.2.20:14999"}) { + t.Fatalf("unauthenticated/relayed address became learned authority: %v", got) + } +} + +func TestRoamingHostileHintCannotRedirectPinnedIdentity(t *testing.T) { + m, b, hostile := newTestManager(t), newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + hits := 0 + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + _ = json.NewEncoder(w).Encode(hostile.buildLocalRoster()) + })) + srv.TLS = hostile.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "https://") + host, port := splitAddr(addr, m.port) + m.upsertMember(&ClusterNode{NodeUUID: b.identity.NodeUUID, ID: "peer", IPAddress: host, Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + outcome, _ := m.reconcileWith([]string{addr}, b.identity.NodeUUID) + if outcome != reconcileUnreachable || hits != 0 { + t.Fatalf("hostile address redirected identity: outcome=%v hits=%d", outcome, hits) + } +} + +func TestRoamingInboundSourceSurvivesRestart(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingRequest(t, m, b, "10.0.0.2:50000", nil) + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + n, _ := m.memberByNodeID(b.identity.NodeUUID) + got := m.resolvePeerAddrs(*n) + if !slices.Equal(got, []string{"10.0.0.2:14999", "192.0.2.20:14999"}) { + t.Fatalf("restart lost source or pairing fallback: %v", got) + } +} diff --git a/services/nvpair-cluster-manager/roster.go b/services/nvpair-cluster-manager/roster.go index d33207a6e..511b33bdd 100644 --- a/services/nvpair-cluster-manager/roster.go +++ b/services/nvpair-cluster-manager/roster.go @@ -36,6 +36,7 @@ type Roster struct { Members []RosterEntry `json:"members"` Tombstones []Tombstone `json:"tombstones,omitempty"` // legacy compatibility RemovalProofs []RemovalProof `json:"removalProofs,omitempty"` + SelfAddresses []string `json:"selfAddresses,omitempty"` // only the authenticated sender, never relayed } // selfPub returns this node's own Ed25519 public key. @@ -59,6 +60,7 @@ func (m *Manager) buildLocalRoster() *Roster { cid, _ := m.clusterIdentity() r := &Roster{ ClusterID: cid, + SelfAddresses: localRosterAddresses(m.port), Tombstones: m.snapshotTombstones(), RemovalProofs: m.snapshotRemovalProofs(), } @@ -107,6 +109,10 @@ func (m *Manager) buildLocalRoster() *Roster { func (m *Manager) mergeRoster(remote *Roster, senderUUID string) bool { m.rosterMu.Lock() defer m.rosterMu.Unlock() + return m.mergeRosterLocked(remote, senderUUID) +} + +func (m *Manager) mergeRosterLocked(remote *Roster, senderUUID string) bool { if m.teardownPending.Load() { return false } @@ -221,7 +227,7 @@ func (m *Manager) applyMembers(entries []RosterEntry, cid, senderUUID string) bo if !m.entryEndorsed(e, cid) { continue } - if err := m.pinEntry(e, der); err != nil { + if err := m.pinEntry(e, der, senderUUID); err != nil { continue } if err := m.clearRemovalProofAfterAdmission(e.NodeUUID, e.AdmissionEpoch); err != nil { @@ -235,7 +241,6 @@ func (m *Manager) applyMembers(entries []RosterEntry, cid, senderUUID string) bo if err := m.trust.AddEndorsements(e.NodeUUID, e.Endorsements); err != nil { log.Printf("roster: persist endorsements for %s: %v", e.NodeUUID, err) } - m.refreshMemberAddr(e) if e.NodeUUID == senderUUID && m.refreshMemberIdentity(e) { changed = true } @@ -251,7 +256,7 @@ func (m *Manager) applyMembers(entries []RosterEntry, cid, senderUUID string) bo if !m.entryEndorsed(e, cid) { continue // no endorsement from a node we currently trust } - if err := m.pinEntry(e, der); err != nil { + if err := m.pinEntry(e, der, senderUUID); err != nil { continue } if err := m.clearRemovalProofAfterAdmission(e.NodeUUID, e.AdmissionEpoch); err != nil { @@ -307,9 +312,15 @@ func (m *Manager) supersededByTombstone(e *RosterEntry) bool { } // pinEntry pins an accepted roster entry's cert and records it as a member. -func (m *Manager) pinEntry(e *RosterEntry, der []byte) error { +func (m *Manager) pinEntry(e *RosterEntry, der []byte, senderUUID string) error { now := time.Now().UnixMilli() - host, port := splitAddr(e.Addr, m.port) + // Endorsements authenticate membership, not a relayed address claim. + host, port := "", m.port + if e.NodeUUID == senderUUID { + if addr, ok := candidateEndpoint(e.Addr, false); ok { + host, port = splitAddr(addr, m.port) + } + } cid, _ := m.clusterIdentity() previous, hadPrevious := m.memberByNodeID(e.NodeUUID) m.upsertMember(&ClusterNode{ @@ -350,25 +361,6 @@ func (m *Manager) pinEntry(e *RosterEntry, der []byte) error { return nil } -// refreshMemberAddr updates a known member's reachable address from a roster -// entry without disturbing its other fields. -// -// A host this node has seen the member connect from outranks the entry's claim, -// which may be the member's own guess about which of its addresses works from -// here — or another peer's relayed copy of that guess. The port is taken from the -// entry either way: a listening port is a fact the member knows and an observer -// cannot infer, since the source port of an inbound connection is ephemeral. -func (m *Manager) refreshMemberAddr(e *RosterEntry) { - if e.Addr == "" { - return - } - host, port := splitAddr(e.Addr, m.port) - if observed := m.observedPeerHost(e.NodeUUID); observed != "" { - host = observed - } - m.setMemberAddr(e.NodeUUID, host, port) -} - // refreshMemberIdentity updates a still-trusted peer's display id/name — in both // the member record and the persisted pin — from its own roster entry, keyed by // the stable nodeUuid. This tracks a peer's PC rename so the roster and UI show diff --git a/services/nvpair-cluster-manager/roster_http.go b/services/nvpair-cluster-manager/roster_http.go index f78425b91..c15db309c 100644 --- a/services/nvpair-cluster-manager/roster_http.go +++ b/services/nvpair-cluster-manager/roster_http.go @@ -10,11 +10,8 @@ import ( "io" "log" "net/http" - "slices" "sync" "time" - - "nvpair-shared/netpick" ) // reconcileEvery is the heartbeat cadence for catch-up roster reconciles — the @@ -43,14 +40,21 @@ func (m *Manager) handleRoster(w http.ResponseWriter, r *http.Request) { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } + // Snapshot the pin and both admissions before reading an untrusted body. + // Removal/rejoin may run while that read blocks; recheck before any merge + // or learning, under the same lock as membership removal. + m.rosterMu.Lock() senderUUID, ok := m.verifyClientPin(r) - if !ok { + pin, pinned := m.trust.Get(senderUUID) + cid0, epoch0 := m.currentAdmission() + m.rosterMu.Unlock() + if !ok || !pinned { m.rejectRosterReconcile(w, r) return } - body, err := io.ReadAll(io.LimitReader(r.Body, maxBodyBytes)) - if err != nil { - http.Error(w, "read body", http.StatusBadRequest) + body, err := io.ReadAll(io.LimitReader(r.Body, maxBodyBytes+1)) + if err != nil || len(body) > maxBodyBytes { + http.Error(w, "invalid body", http.StatusBadRequest) return } var remote Roster @@ -58,20 +62,25 @@ func (m *Manager) handleRoster(w http.ResponseWriter, r *http.Request) { http.Error(w, "invalid JSON", http.StatusBadRequest) return } - changed := m.mergeRoster(&remote, senderUUID) + m.rosterMu.Lock() + cid, epoch := m.currentAdmission() + current, stillPinned := m.trust.Get(senderUUID) + _, valid := m.verifyClientPin(r) + if !valid || !stillPinned || current.AdmissionEpoch != pin.AdmissionEpoch || cid == "" || cid != cid0 || epoch != epoch0 { + m.rosterMu.Unlock() + http.Error(w, "stale peer", http.StatusForbidden) + return + } + changed := m.mergeRosterLocked(&remote, senderUUID) + if remote.ClusterID == cid { + changed = m.learnPeerAddressesLocked(senderUUID, hostOnly(r.RemoteAddr), remote.SelfAddresses, "", time.Now()) || changed + } + // Persist under the composition boundary: an old in-flight request must + // never write learned addresses back after member removal. if changed { - log.Printf("roster: reconciled from peer %s; membership changed", senderUUID) - } - // Learn the peer's current address from the authenticated connection so a - // member that moved to a new IP becomes reachable again without waiting on - // mDNS. verifyClientPin already authenticated the sender against its pinned - // cert, so the TCP source IP is trustworthy enough to dial back on the - // member's known listening port (we pass port 0 to keep the stored - // listening port, never the ephemeral source port). - if host := hostOnly(r.RemoteAddr); host != "" && m.noteObservedPeerHost(senderUUID, host) { - log.Printf("roster: refreshed peer %s address to %s (from connection)", senderUUID, host) - changed = true + m.persistMembers() } + m.rosterMu.Unlock() if changed { m.emitNodesChanged() } @@ -214,6 +223,7 @@ func (m *Manager) reconcileWith(addrs []string, peerUUID string) (reconcileOutco // reconcileWithin is reconcileWith under a caller-supplied budget. func (m *Manager) reconcileWithin(ctx context.Context, addrs []string, peerUUID string) (reconcileOutcome, bool) { + addrs = m.reservePeerDialCandidates(peerUUID, addrs) if len(addrs) == 0 { return reconcileUnreachable, false } @@ -265,6 +275,13 @@ func confirmedFirst(addrs []string, confirmed string) []string { // reconcileWith to try the next candidate; every actual response — including a // rejection — is that peer's answer and ends the walk. func (m *Manager) reconcileOnce(ctx context.Context, client *http.Client, body []byte, addr, peerUUID string) (outcome reconcileOutcome, provenRemoval, answered bool) { + m.rosterMu.Lock() + cid0, epoch0 := m.currentAdmission() + pin0, pinned := m.trust.Get(peerUUID) + m.rosterMu.Unlock() + if !pinned { + return reconcileUnreachable, false, false + } req, err := http.NewRequestWithContext(ctx, http.MethodPost, peerURL("https", addr, rosterPath), bytes.NewReader(body)) if err != nil { log.Printf("roster: build reconcile request for %s (%s): %v", addr, peerUUID, err) @@ -277,7 +294,17 @@ func (m *Manager) reconcileOnce(ctx context.Context, client *http.Client, body [ return reconcileUnreachable, false, false } defer resp.Body.Close() - rb, _ := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes)) + rb, readErr := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes+1)) + m.rosterMu.Lock() + defer m.rosterMu.Unlock() + cid, epoch := m.currentAdmission() + pin, stillPinned := m.trust.Get(peerUUID) + if !stillPinned || pin.AdmissionEpoch != pin0.AdmissionEpoch || cid == "" || cid != cid0 || epoch != epoch0 || m.teardownPending.Load() || resp.TLS == nil || len(resp.TLS.PeerCertificates) == 0 || !m.trust.MatchDER(peerUUID, resp.TLS.PeerCertificates[0].Raw) { + return reconcileUnreachable, false, false + } + if readErr != nil || len(rb) > maxBodyBytes { + return reconcileUnreachable, false, true + } // A 403 is an authenticated signal (we pinned the peer's server cert, so we // know it is really that peer) that it no longer holds our pin. But a dropped // pin alone does not prove *we* were removed — a peer that itself left drops @@ -302,8 +329,11 @@ func (m *Manager) reconcileOnce(ctx context.Context, client *http.Client, body [ log.Printf("roster: decode reconcile response from %s: %v", peerUUID, err) return reconcileAccepted, false, true // 200: the peer accepted us; only the response body was unusable } - if m.mergeRoster(&remote, peerUUID) { - log.Printf("roster: reconciled with peer %s; membership changed", peerUUID) + changed := m.mergeRosterLocked(&remote, peerUUID) + if remote.ClusterID == cid { + changed = m.learnPeerAddressesLocked(peerUUID, "", remote.SelfAddresses, addr, time.Now()) || changed + } + if changed { m.emitNodesChanged() } return reconcileAccepted, false, true @@ -467,12 +497,8 @@ func (m *Manager) removeBareRejectingPeers(clusterID string, generation uint64, } } -// reconcileLoop runs the periodic catch-up reconcile and tombstone GC until the -// context is cancelled. Each pass first refreshes member addresses from the -// mDNS browse map, so a member that moved to a new IP (sleep/wake, DHCP) -// re-converges within a heartbeat without a dedicated discovery callback. An -// initial pass shortly after startup converges a node that just (re)started or -// woke at a new address without waiting a full heartbeat. +// reconcileLoop runs catch-up reconciliation and tombstone GC until cancelled. +// Only pinned roster connections can update a member's durable address hints. func (m *Manager) reconcileLoop(ctx context.Context) { ticker := time.NewTicker(reconcileEvery) defer ticker.Stop() @@ -490,8 +516,8 @@ func (m *Manager) reconcileLoop(ctx context.Context) { } } -// reconcilePass GCs expired tombstones, refreshes member addresses from the -// discovery resolver, and pushes a roster reconcile to every confirmed peer. +// reconcilePass GCs expired tombstones and pushes a roster reconcile to every +// confirmed peer, using its bounded authenticated address hints. func (m *Manager) reconcilePass() { m.gcTombstones() if cid, _ := m.clusterIdentity(); cid == "" { @@ -499,98 +525,11 @@ func (m *Manager) reconcilePass() { } m.mesh.Refresh() m.clients.DropUnpinned() - m.refreshMemberAddrsFromMDNS() m.reconcilePeersAndMaybeSelfRemove() } -// refreshMemberAddrsFromMDNS updates each confirmed peer's stored reachable -// address from the mDNS browse map when the peer has stopped advertising the -// address we hold. Inter-node calls are mTLS cert-pinned, so trusting mDNS only -// to *locate* a peer is safe — a wrong address simply fails the pinned handshake -// and the next pass corrects it. This is the backstop that re-converges a moved -// member even when it can't reach us to announce its new address itself. -// -// A peer that still advertises the address we hold keeps it. That address was -// observed working from here, whereas the peer's own first choice is ranked from -// its vantage point and may be a link only its cabled neighbour can use, so -// overwriting on every pass would trade a known-good endpoint for a guess. -func (m *Manager) refreshMemberAddrsFromMDNS() { - if m.browser == nil { - return - } - changed := false - for _, n := range m.snapshotNodes() { - if n.NodeUUID == m.identity.NodeUUID || n.State != stateMember { - continue - } - hosts, port, ok := m.browser.Resolve(n.NodeUUID) - if !ok { - hosts, port, ok = m.browser.Resolve(n.ID) - } - if !ok || len(hosts) == 0 { - continue - } - host := hosts[0] - if slices.Contains(hosts, n.IPAddress) { - // A peer that still advertises what we hold keeps it, unless hosts[0] is - // a strictly higher-ranked address class (e.g. physical private LAN over - // Tailscale / CGNAT). This allows seamless upgrade to local Wi-Fi when - // reconnecting to the local network. - if netpick.ScoreAddress(hosts[0]) <= netpick.ScoreAddress(n.IPAddress) { - host = n.IPAddress - } - } - if m.setMemberAddr(n.NodeUUID, host, port) { - log.Printf("roster: refreshed peer %s address to %s (from discovery)", n.NodeUUID, joinHostPort(host, port, m.port)) - changed = true - } - } - if changed { - m.emitNodesChanged() - } -} - -// resolvePeerAddrs returns a member's host:port candidates in the order to try -// them: the address recorded at pairing time first (authoritative — it is the -// source IP the peer connected from plus the listening port it advertised), then -// everything the peer currently advertises through mDNS. -// -// The recorded address leads on equal footing, but if it is a lower-ranked class -// (e.g. CGNAT/Tailscale) than an available physical LAN address from discovery, -// discovery's top candidate leads so local Wi-Fi is preferred when reachable. +// resolvePeerAddrs uses only local, authenticated address hints. Discovery and +// relayed names/addresses cannot modify this set. func (m *Manager) resolvePeerAddrs(n ClusterNode) []string { - addrs := make([]string, 0, 4) - seen := make(map[string]bool, 4) - add := func(addr string) { - if addr == "" || seen[addr] { - return - } - seen[addr] = true - addrs = append(addrs, addr) - } - var hosts []string - var port int - if m.browser != nil { - var ok bool - hosts, port, ok = m.browser.Resolve(n.NodeUUID) - if !ok { - hosts, port, ok = m.browser.Resolve(n.ID) - } - } - leadWithRecorded := n.IPAddress != "" - if leadWithRecorded && len(hosts) > 0 { - if netpick.ScoreAddress(hosts[0]) > netpick.ScoreAddress(n.IPAddress) { - leadWithRecorded = false - } - } - if leadWithRecorded { - add(joinHostPort(n.IPAddress, n.Port, m.port)) - } - for _, h := range hosts { - add(joinHostPort(h, port, m.port)) - } - if !leadWithRecorded && n.IPAddress != "" { - add(joinHostPort(n.IPAddress, n.Port, m.port)) - } - return addrs + return memberCandidateAddrs(n, time.Now()) } diff --git a/services/nvpair-manual-nodes/README.md b/services/nvpair-manual-nodes/README.md index 4358b7eec..d110cf94b 100644 --- a/services/nvpair-manual-nodes/README.md +++ b/services/nvpair-manual-nodes/README.md @@ -11,6 +11,11 @@ Node-info JSON decoding reads at most 1 MiB, and model-list decoding at most 16 MiB. A JSON value that extends past its limit follows the existing probe or model-parse failure path. +Cluster-derived entries follow cluster-manager's authenticated reached address +when a member roams. User-saved PAIR/engine entries retain their original +address, name and intent; the learned roster never overwrites them. Each new +endpoint still needs a pinned-mTLS inventory proof before it becomes routable. + ## Worker control health An id-bearing `worker/health` request (no params) returns `{"ok":true}` with diff --git a/services/nvpair-manual-nodes/manager.go b/services/nvpair-manual-nodes/manager.go index e7f3c3bef..5880523a4 100644 --- a/services/nvpair-manual-nodes/manager.go +++ b/services/nvpair-manual-nodes/manager.go @@ -577,8 +577,10 @@ func (m *Manager) syncClusterMembers() { } found := false for _, member := range activeMembers { - if (tn.status.HostUUID != "" && tn.status.HostUUID == member.NodeUUID) || - tn.entry.Address == member.IPAddress { + // A cluster seed follows the member's reached endpoint. A saved + // entry is never rebuilt here, so its operator-selected address wins. + if tn.entry.Address == member.IPAddress && + (tn.status.HostUUID == "" || tn.status.HostUUID == member.NodeUUID) { found = true break } diff --git a/services/nvpair-manual-nodes/roaming_test.go b/services/nvpair-manual-nodes/roaming_test.go new file mode 100644 index 000000000..f7c7aa2ec --- /dev/null +++ b/services/nvpair-manual-nodes/roaming_test.go @@ -0,0 +1,63 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" +) + +func TestClusterSeedFollowsRoamingAddressWithoutOverwritingSavedEntry(t *testing.T) { + for _, saved := range []bool{false, true} { + name := "cluster seed" + if saved { + name = "saved alias" + } + t.Run(name, func(t *testing.T) { + m, _, _ := newTestManager(t) + m.clusterDir = t.TempDir() + const uuid = "22222222-2222-2222-2222-222222222222" + write := func(addr string, tick int) { + t.Helper() + data, err := json.Marshal([]clusterMemberEntry{{ID: "peer", Name: "peer", NodeUUID: uuid, IPAddress: addr, State: "member"}}) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(m.clusterDir, "members.json") + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + at := time.Now().Add(time.Duration(tick) * time.Second) + if err := os.Chtimes(path, at, at); err != nil { + t.Fatal(err) + } + m.syncClusterMembers() + } + // Start with a previously proven tracked peer, without a background + // probe racing the address-change interleaving being tested. + entry := ManualEntry{Kind: "pair", Address: "192.0.2.20", Name: "peer"} + if saved { + entry.Name = "my-saved-peer" + } + m.mu.Lock() + m.nodes[nodeID(entry)] = &trackedNode{entry: entry, fromCluster: !saved, status: ManualNodeStatus{ID: nodeID(entry), Kind: "pair", Address: entry.Address, HostUUID: uuid, ClusterUUID: uuid, FromCluster: !saved}} + m.mu.Unlock() + write("100.64.0.8", 2) + nodes := m.listNodes() + if len(nodes) != 1 { + t.Fatalf("nodes=%+v", nodes) + } + want := "100.64.0.8" + if saved { + want = "192.0.2.20" + } + if nodes[0].Address != want { + t.Fatalf("reached address=%q want %q", nodes[0].Address, want) + } + }) + } +} diff --git a/services/nvpair-node-scanner/README.md b/services/nvpair-node-scanner/README.md index 3e00e89e8..5bb208868 100644 --- a/services/nvpair-node-scanner/README.md +++ b/services/nvpair-node-scanner/README.md @@ -9,6 +9,12 @@ SPDX-License-Identifier: Apache-2.0 Concretely, the daemon: +- **Waits for local node-info on startup:** only the first local inventory read + retries with the existing 2, 4, 8-second telemetry backoff, under a single + 15-second request-and-wait budget. A normal listener startup logs no outage; + a real outage still warns when that budget ends. After the first successful + read (or exhausted startup budget), failures warn immediately as before. + - **Advertises** this node's one `_nvpair-node._tcp` record, built from a registry of the local services its parent (the broker) registers — `ni`/`ol`/`lm`/`er`/`wl`/`cl`/`em`/`ec` ports plus the node identity (`uuid=`, `cluster-uuid=` when clustered, `ip=`). One record per node covers every service. - **Browses** `_nvpair-node._tcp` for candidate endpoints. Verified/local records are keyed by authenticated host UUID; unverified previews have independent `preview:` record keys and an explicit `identityVerification: "unverified"` marker. A claimed UUID cannot overwrite or remove a verified record. - **Authenticates before enrichment:** remote hardware and broker-owned service metadata come from `ni` over pinned cluster mTLS, with matching certificate/response/metadata identities and an exact-current-pin check after reading. Only authenticated metadata supplies service ports. Models, per-engine attribution and loaded state then come from the authenticated `em` port over pinned mTLS. Local inventory remains an owned loopback read. Public HTTP hardware is display-only and never supplies service models or scheduler telemetry. diff --git a/services/nvpair-node-scanner/daemon.go b/services/nvpair-node-scanner/daemon.go index 789ced50f..71502a6cf 100644 --- a/services/nvpair-node-scanner/daemon.go +++ b/services/nvpair-node-scanner/daemon.go @@ -100,6 +100,8 @@ type daemon struct { // into a torn self entry. The self key is written nowhere else — onBrowse // ignores our own uuid — so this is the sole self-key writer lock. selfMu sync.Mutex + // Only the first local enrichment waits for node-info's startup readiness. + localInfoStartup sync.Once // lastInfo / lastModels cache each node's last successful enrichment (keyed by // hostUuid) — GPU/CPU/memory from node-info and the model list from @@ -1231,7 +1233,7 @@ func (d *daemon) enrichInfoCandidates(node *noderec.DirectoryNode, hosts []strin return } ask := func(host string) (NodeInfoResponse, bool) { - candidate, fetched := d.fetchNodeInfo(host, ni.Port) + candidate, fetched := d.fetchLocalNodeInfo(host, ni.Port, node.HostUUID) if !fetched { return NodeInfoResponse{}, false } diff --git a/services/nvpair-node-scanner/local_readiness.go b/services/nvpair-node-scanner/local_readiness.go new file mode 100644 index 000000000..5f5b89d36 --- /dev/null +++ b/services/nvpair-node-scanner/local_readiness.go @@ -0,0 +1,47 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "time" +) + +const localNodeInfoStartupTimeout = 15 * time.Second + +// fetchLocalNodeInfo gives only the first local inventory read a readiness +// grace. The broker spawns scanner before node-info's listener is ready. Reuse +// the existing telemetry backoff, bounded across requests and waits, so normal +// startup is quiet but a real outage still reaches noteNodeInfo's warning. +// Later outages retain the immediate transition report and last-good cache. +func (d *daemon) fetchLocalNodeInfo(host string, port int, expectedUUID string) (NodeInfoResponse, bool) { + var info NodeInfoResponse + var ok, initial bool + d.localInfoStartup.Do(func() { + initial = true + if d.http == nil && d.tlsHTTP == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), localNodeInfoStartupTimeout) + defer cancel() + for attempt := 0; ctx.Err() == nil; attempt++ { + info, ok = d.fetchNodeInfoWithin(ctx, host, port) + if ok && (info.HostUUID == "" || info.HostUUID == expectedUUID) { + return + } + ok = false + timer := time.NewTimer(telemetryRetryDelay(attempt)) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } + }) + if initial { + return info, ok + } + return d.fetchNodeInfo(host, port) +} diff --git a/services/nvpair-node-scanner/local_readiness_test.go b/services/nvpair-node-scanner/local_readiness_test.go new file mode 100644 index 000000000..1a36558f1 --- /dev/null +++ b/services/nvpair-node-scanner/local_readiness_test.go @@ -0,0 +1,84 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "nvpair-shared/noderec" +) + +func localReadinessFixture(t *testing.T, handler http.Handler) (*daemon, *noderec.DirectoryNode) { + t.Helper() + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + u, err := url.Parse(srv.URL) + if err != nil { + t.Fatal(err) + } + port, err := strconv.Atoi(u.Port()) + if err != nil { + t.Fatal(err) + } + d := newSelfTestDaemon("self", "127.0.0.1") + return d, &noderec.DirectoryNode{HostUUID: "self", Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceNodeInfo: {Port: port}}} +} + +func TestLocalNodeInfoStartupWaitsBeforeWarning(t *testing.T) { + logs := captureLogs(t) + var requests atomic.Int32 + d, node := localReadinessFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if requests.Add(1) == 1 { + http.Error(w, "starting", http.StatusServiceUnavailable) + return + } + _ = json.NewEncoder(w).Encode(NodeInfoResponse{HostUUID: "self"}) + })) + d.enrichInfoAt(node, "127.0.0.1") + if requests.Load() < 2 { + t.Fatal("startup did not retry local node-info readiness") + } + if _, ok := d.lastInfo["self"]; !ok { + t.Fatal("local inventory not read after startup") + } + if strings.Contains(logs.String(), "not answering") { + t.Fatalf("normal startup warned: %s", logs.String()) + } + // Once it answered, a later outage must warn immediately, not get a new grace. + d.http.CloseIdleConnections() + d.http = &http.Client{Transport: readinessUnavailable{}} + start := time.Now() + d.enrichInfoAt(node, "127.0.0.1") + if !strings.Contains(logs.String(), "not answering") || time.Since(start) > time.Second { + t.Fatalf("later outage was hidden/delayed: %s", logs.String()) + } +} + +type readinessUnavailable struct{} + +func (readinessUnavailable) RoundTrip(*http.Request) (*http.Response, error) { + return nil, http.ErrServerClosed +} + +func TestLocalNodeInfoRealStartupOutageEventuallyWarns(t *testing.T) { + logs := captureLogs(t) + d, node := localReadinessFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.Error(w, "down", 503) })) + start := time.Now() + d.enrichInfoAt(node, "127.0.0.1") + elapsed := time.Since(start) + if elapsed < 10*time.Second || elapsed > 20*time.Second { + t.Fatalf("startup readiness wait=%v, want bounded grace", elapsed) + } + if got := strings.Count(logs.String(), "not answering"); got != 1 { + t.Fatalf("outage warnings=%d: %s", got, logs.String()) + } +} diff --git a/services/tests/peer_roaming_test.go b/services/tests/peer_roaming_test.go new file mode 100644 index 000000000..3a9a2b457 --- /dev/null +++ b/services/tests/peer_roaming_test.go @@ -0,0 +1,105 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package tests + +import ( + "encoding/json" + "os" + "path/filepath" + "strconv" + "testing" + "time" +) + +// Real isolated workers: corrupt A's old address for B, let B's pinned inbound +// roster connection repair it, restart A, then prove A reaches B by delivering +// authenticated removal. No multicast prerequisite or environment skip. +func TestPeerRoamingLearnsInboundAndDialsAfterRestart(t *testing.T) { + dirA, dirB := t.TempDir(), t.TempDir() + portA, portB := freePort(t), freePort(t) + a, b := startCM(t, dirA, portA), startCM(t, dirB, portB) + defer func() { + if a != nil { + a.stop() + } + if b != nil { + b.stop() + } + }() + waitForPort(t, "127.0.0.1", portA, 5*time.Second) + waitForPort(t, "127.0.0.1", portB, 5*time.Second) + pairNodes(t, a, b, portB, "peer-b") + bid := decodeResult[cmNodeID](t, b.call("cluster:get-node-id", nil)).NodeUUID + a.stop() + a = nil + b.stop() + b = nil + type diskMember struct { + NodeUUID string `json:"nodeUuid"` + IPAddress string `json:"ipAddress"` + PairingAddress string `json:"pairingAddress"` + AddressCandidates []struct { + Address string `json:"address"` + } `json:"addressCandidates"` + } + path := filepath.Join(dirA, "cluster", "members.json") + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var raw []map[string]json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + t.Fatal(err) + } + stale := "192.0.2.20:" + strconv.Itoa(portB) + for _, member := range raw { + var uuid string + _ = json.Unmarshal(member["nodeUuid"], &uuid) + if uuid == bid { + member["ipAddress"], _ = json.Marshal("192.0.2.20") + member["pairingAddress"], _ = json.Marshal(stale) + delete(member, "addressCandidates") + } + } + data, _ = json.Marshal(raw) + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + a = startCM(t, dirA, portA) + b = startCM(t, dirB, portB) + reached := false + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + data, err = os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var members []diskMember + if err := json.Unmarshal(data, &members); err != nil { + t.Fatal(err) + } + for _, member := range members { + if member.NodeUUID != bid || member.IPAddress != "127.0.0.1" { + continue + } + if member.PairingAddress != stale || len(member.AddressCandidates) == 0 { + t.Fatalf("repair lost pairing fallback or durable learned source: %+v", member) + } + reached = true + } + if reached { + break + } + time.Sleep(100 * time.Millisecond) + } + a.stop() + a = nil + if !reached { + t.Fatal("B's authenticated inbound connection never repaired A's stale member address") + } + a = startCM(t, dirA, portA) + waitForPort(t, "127.0.0.1", portA, 5*time.Second) + a.call("nodes:remove", map[string]string{"nodeUuid": bid}) + waitMembers(t, b, 0) +} diff --git a/services/versions.json b/services/versions.json index 2f5c0cfe3..0a255a572 100644 --- a/services/versions.json +++ b/services/versions.json @@ -6,14 +6,14 @@ "ollama-proxy": "1.5.0", "lmstudio-proxy": "1.5.0", "nvpair-node-info": "1.4.0", - "nvpair-node-scanner": "1.3.0", - "nvpair-manual-nodes": "1.3.0", + "nvpair-node-scanner": "1.3.1", + "nvpair-manual-nodes": "1.3.1", "nvpair-workload-manager": "1.3.1", "nvpair-errors": "1.1.1", "nvpair-node-settings": "3.2.0", "nvpair-ui-broker": "2.5.0", "nvpair-engine-manager": "1.4.0", - "nvpair-cluster-manager": "2.2.0", + "nvpair-cluster-manager": "2.3.0", "nvpair-job-scheduler": "1.1.0", "nvpair-tui": "1.3.0", "omlx-proxy": "1.5.0"