Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions docs/architecture.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
42 changes: 42 additions & 0 deletions services/nvpair-cluster-manager/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
283 changes: 283 additions & 0 deletions services/nvpair-cluster-manager/addresses.go
Original file line number Diff line number Diff line change
@@ -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
}
Loading
Loading